package cn.doitedu.profile;

import cn.doitedu.eagle.beans.LogBean;
import cn.doitedu.demo.LogBeanFlat;
import cn.doitedu.profile.utils.LogBeanUtil;
import com.alibaba.fastjson.JSON;
//import com.dorisdb.connector.flink.DorisSink;
//import com.dorisdb.connector.flink.table.DorisSinkOptions;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/***
 * @author hunter.d
 * @qq 657270652
 * @wx haitao-duan
 * @date 2021/3/13
 **/
public class EventsLoad2Doris {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        Properties props = new Properties();

        props.setProperty("bootstrap.servers", "hdp01:9092,hdp02:9092,hdp03:9092");
        props.setProperty("auto.offset.reset", "earliest");
        FlinkKafkaConsumer<String> kafkaSource
                = new FlinkKafkaConsumer<String>("app_log", new SimpleStringSchema(), props);

        DataStreamSource<String> ds = env.addSource(kafkaSource);

        SingleOutputStreamOperator<String> beanStrea0m = ds.map(new MapFunction<String, String>() {
            @Override
            public String map(String value) throws Exception {
                LogBean logBean = JSON.parseObject(value, LogBean.class);
                LogBeanFlat flatBean = LogBeanUtil.flatBean(logBean);
                return JSON.toJSONString(flatBean,false);
            }
        });

        /*beanStream.addSink(
                DorisSink.sink(
                        // the sink options
                        DorisSinkOptions.builder()
                                .withProperty("jdbc-url", "jdbc:mysql://hdp01:9030")
                                .withProperty("load-url", "hdp01:8030")
                                .withProperty("username", "root")
                                .withProperty("password", "root")
                                .withProperty("table-name", "event_detail")
                                .withProperty("database-name", "xdb")
                                .withProperty("sink.properties.format", "json")
                                .withProperty("sink.properties.strip_outer_array", "true")
                                .build()
                )
        );

        beanStream.print();*/
        env.execute();


    }
}
